package com.atguigu.mr.topn;

import com.atguigu.mr.flow.FlowBean;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.Set;
import java.util.TreeMap;

/**
 * @description: 进行分区-在分区中又进行排序
 * @time: 20191226 18:03
 * @author: baojinlong
 **/
public class TopNMapper extends Mapper<LongWritable, Text, FlowBean, Text> {

    // 定义一个TreeMap作为存储数据的容器（天然按key排序）
    private TreeMap<FlowBean, Text> flowMap = new TreeMap<FlowBean, Text>();
    private FlowBean flowBean = new FlowBean();
    private Text v = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 1 获取一行
        String line = value.toString();
        // 2 截取
        String[] fields = line.split("\t");
        // 3 封装对象
        String phoneNbr = fields[0];
        long upFlow = Long.parseLong(fields[1]);
        flowBean.setUpFlow(upFlow);
        long downFlow = Long.parseLong(fields[2]);
        flowBean.setDownFlow(downFlow);
        long sumFlow = Long.parseLong(fields[3]);
        flowBean.setSumFlow(sumFlow);
        v.set(phoneNbr);

        // 4 向TreeMap中添加数据
        flowMap.put(flowBean, v);
        // 5 限制TreeMap的数据量，超过10条就删除掉流量最小的一条数据
        if (flowMap.size() > 10) {
            // flowMap.remove(flowMap.firstKey());
            flowMap.remove(flowMap.lastKey());
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        // 6 遍历treeMap集合，输出数据
        Set<FlowBean> flowBeans = flowMap.keySet();
        for (FlowBean k : flowBeans) {
            // 4 输出
            context.write(k, flowMap.get(k));
        }
    }
}
